package com.ctrip.corp.obt.shard.manager;


import com.ctrip.corp.obt.generic.utils.CatUtils;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;
import com.ctrip.corp.obt.generic.utils.StringUtils;
import com.ctrip.corp.obt.shard.core.DataPlaneSource;
import com.ctrip.corp.obt.shard.holder.ShardSourceHolder;
import com.ctrip.corp.obt.shard.loader.IShardSourceLoader;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Message;
import com.google.common.collect.Sets;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.Set;

/**
 * @author marco.zheng
 * @version 1.0
 * @date 2022/03/14
 * @desc 数据面加载源manager
 * @see
 * @since 1.0
 */
@Slf4j
@RequiredArgsConstructor
public class DataPlaneSourceManager {

    private final IShardSourceLoader defaultShardSourceLoader;

    public Set<String> loadBySources(DataPlaneSource dataPlaneSource) {
        Set<String> tenantIds = dataPlaneSource.getTenantIds();
        String source = dataPlaneSource.getSource();
        if (CollectionUtils.isEmpty(tenantIds)) {
            return Collections.emptySet();
        }
        return CatUtils.wrapEvent(defaultShardSourceLoader::loadBySources, tenantIds,
                sourceList -> {
                    Sets.SetView<String> difference = Sets.difference(tenantIds, sourceList);
                    if (log.isDebugEnabled()) {
                        log.debug("Success data plane loading source, source tenant ids: {}", sourceList);
                    }
                    Cat.logMetricForCount(StringUtils.join("DataShard:Load.Source:", source));
                    if (CollectionUtils.isNotEmpty(difference)) {
                        if (log.isDebugEnabled()) {
                            log.debug("Failed Data plane loading source, source tenant ids: {}", tenantIds);
                        }
                        Cat.logEvent("DataShard", "Load.Source", Message.SUCCESS, StringUtils.join("failed for loading tenantIds:", tenantIds));
                    }
                }, throwable -> {
                    log.error("Loading source failed, tenant ids: {}, reason is {}", tenantIds, throwable.getMessage());
                    Cat.logEvent("DataShard", "Load.Source", Message.SUCCESS, StringUtils.join("failed for loading tenantIds:", tenantIds));
                    throwable.printStackTrace();
                    return Collections.emptySet();
                });
    }

    public Set<String> listSourceMembers() {
        return ShardSourceHolder.getDataSourceMembers();
    }
}
